package com.gugee.ins.data.analysis.listener;

import com.gugee.ins.data.analysis.config.InsAnalysisConfig;
import com.gugee.ins.data.analysis.service.InsAnalysisAvatarService;
import com.gugee.ins.data.analysis.service.InsAnalysisIncomeService;
import com.gugee.ins.data.common.consumer.AbstractGugeeConsumerHandler;
import com.gugee.ins.data.common.msg.InsBloggerMessage;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.List;


/**
 * InsAnalysisAvatar Listener
 *
 * @author Huangxin
 */
@Component
@Slf4j
public class InsAnalysisAvatarListener {

    /**
     * InsBloggerFollowingConfig
     */
    @Autowired
    private InsAnalysisConfig insAnalysisConfig;

    /**
     * insBloggerFollowerService
     */
    @Autowired
    private InsAnalysisAvatarService insAnalysisAvatarService;

    /**
     * insAnalysisInComeService
     */
    @Autowired
    private InsAnalysisIncomeService insAnalysisInComeService;

    /**
     *
     */
    private InsAnalysisAvatarHandler analysisHandler = new InsAnalysisAvatarHandler();

    /**
     *
     */
    private InsAnalysisIncomeHandler analysisIncomeHandler = new InsAnalysisIncomeHandler();

    /**
     * 监听任务,完成头像分析
     * @param messages
     * @param ack
     */
    @KafkaListener(id="${kafka.consumer.group.insAnalysisAvatar.id}",
            topics = {"${kafka.topics.insAnalysisAvatar.online}","${kafka.topics.insAnalysisAvatar.priority}"},
            containerFactory = "factoryInsAnalysisAvatarAck")
    public void insAnalysisAvatar(List<InsBloggerMessage> messages, Acknowledgment ack) {
        analysisHandler.handle("InsAnalysisAvatar",messages, insAnalysisConfig.getAcceptInsAnalysisAvatarMsgTimeMinRange(),ack);
    }

    /**
     * 监听任务,完成收入分析
     * @param messages
     * @param ack
     */
    @KafkaListener(id="${kafka.consumer.group.insAnalysisIncome.id}",
            topics = {"${kafka.topics.insAnalysisIncome.online}"},
            containerFactory = "factoryInsAnalysisIncomeAck")
    public void insAnalysisIncome(List<InsBloggerMessage> messages, Acknowledgment ack) {
        analysisIncomeHandler.handle("InsAnalysisIncome",messages, insAnalysisConfig.getAcceptInsAnalysisIncomeMsgTimeMinRange(),ack);
    }

    class InsAnalysisAvatarHandler extends AbstractGugeeConsumerHandler<InsBloggerMessage>{
        @Override
        public void dealMsg(List<InsBloggerMessage> messages) {
            insAnalysisAvatarService.handleAnalysisAvatar(messages);
        }
    }

    class InsAnalysisIncomeHandler extends AbstractGugeeConsumerHandler<InsBloggerMessage>{
        @Override
        public void dealMsg(List<InsBloggerMessage> messages) {
            insAnalysisInComeService.handleAnalysisIncome(messages);
        }
    }

}
